<?php

namespace app\command;

use app\api\controller\wap\Trade;
use app\common\model\coin\CallApi;
use app\common\model\coin\CoinRuleRun;
use app\common\model\coin\CoinRuleRunTrade;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\Exception;
use think\facade\Db;


/**
 * 获取历史火币数据，并进行保存
 * Class HistoryTrade
 * @package app\command
 */
class MonitorTrade extends Command
{
    /**
     * 初始化的内存表
     */
    const TABLE_LIST = [
        'CONFIG'   => ['config',true,false,1,[      //【表名,是否初始化创建,是否启用锁,记录数（1024的倍数）,字段参数】
            ['value','TYPE_STRING',20],  // 配置的值
        ]], //配置信息
        'API_REQUEST_RUN' => ['api_request_run',true,true,10,[  //键为ID加版本号 id_version
            ['last_time','TYPE_INT',11],  // 最后请求时间
            ['details','TYPE_STRING',2048], //内容
            ['status','TYPE_INT',1],    // 状态：1=>需申请请求；2=>获取数据，待处理，3=>处理中，4=>处理成功，但未修改，5=>完成，6=>请求失败
        ]],  // API 请求参数设置

    ];

    /**
     * 配置表的键列表
     */
    const CONFIG_TABLE_KEY_LIST = [
        'TRADE_REFRESH_CONFIG_KEY'  => 'trade_refresh',  // 判断是否刷新订单状态的键
    ];

    /**
     * 进程参数设置
     */
    const PROCESS_CONFIG = [
        'TRADE_REQUEST_NUM' => 10,  //订单请求数量
        'API_STATUS_WAIT'   => 2,   //待处理，但还未处理
        'API_STATUS_PROCESS'    => 3,   //处理中等待时间
        'API_STATUS_NOT_UDPATE' => 2,   //请求，但未发生变化
        'API_STATUS_FINISH' => 0,   //请求，并处理成功
        'API_STATUS_ERROR'  => 10,   //请求，并失败了

        'DEDUCT_MAX_REQUEST'    => 10000,   // 最大请求次数
        'DEDUCT_ONE_QUERY_MAX'  => 100,     // 每次查询的数量
        'DEDUCT_ERROR_LOG_TYPE' => 'DEDUCT_ERROR',  //折扣获取错误日志类型
    ];

    /**
     * ENV 配置参数KEY列表
     */
    const ENV_KEY_LIST = [
        'TCP_PORT'   => 'monitor_trade.tcp_port',
    ];

    /**
     * 间隔时间执行，单位秒
     */
    const INTERVAL_TIME_LIST = [
        'TRADE_STATUS_REFRESH'  => 20, // 订单监控刷新时间
        'API_STATUS_WAIT'   => 2,   //待处理，但还未处理
        'API_STATUS_PROCESS'    => 3,   //处理中等待时间
        'API_STATUS_NOT_UDPATE' => 100,   //请求，但未发生变化
        'API_STATUS_FINISH' => 0,   //请求，并处理成功
        'API_STATUS_ERROR'  => 25,   //请求，并失败了

        'DEDUCT_TIME'   => 3,  // 抵扣监控间隔时间
    ];

    /**
     * 休眠等待时间，单位微妙
     */
    const USLEEP_TIME_LIST = [
        'TRADE_STATUS'  => 100000, //订单状态监控
        'TRADE_MONITOR_MASTER'  => 400000, //订单监控主进程

        'ERROR_RESTART' => 500000,  //进程失败重启时间

    ];


    /**
     * 重写configure
     * {@inheritdoc}
     */
    protected function configure()
    {
        // 命令的名字（"think" 后面的部分）
        $this->setName('MonitorTrade')
            // 配置一个参数 使用$input->getArgument('username')获取
            // ->addArgument('username')
            // 运行 "php think list" 时的简短描述
            ->setDescription('监控订单状态处理')
            // 运行命令时使用 "--help" 选项时的完整命令描述
            ->setHelp("一直执行 无参数");
    }


    /**
     *  * 重写execute
     *  * {@inheritdoc}
     *
     * @param Input $input
     * @param Output $output
     */
    public function execute(Input $input, Output $output)
    {

        // 创建内存表
        static::swooleTable();

        // 监控订单状态，并将其写入内存中进行处理
        $this->monitorTradeStatus();

        // 订单监控管理进程
        $this->monitorTradeMaster();

        // 开启交互监听
        $this->createTcpServer();

        // 监控抵扣
        $this->monitorDeduct();


        //开启等待
        Swoole::wait();

    }


    /**
     * 获取tcp的监控端口
     */
    static function tcpPort()
    {
        if(!$port = env(static::ENV_KEY_LIST['TCP_PORT'])) {
            $port = 9601;
        }
        return $port;
    }

    /**
     * 失败重启执行函数
     * @param $function
     */
    private function startProcessAndErrorRestart($function,$isMaster = false) {
        Swoole::startProcess(function() use($function){
            while (true) {
                try {
                    // 执行数据处理
                    call_user_func_array($function,[]);
                } catch (\Exception $e) {
             //       echo $e->getMessage(),':',$e->getLine(),"\n";
                    usleep(self::USLEEP_TIME_LIST['ERROR_RESTART']);
                }
            }
        },true,$isMaster);
    }

    /**
     * 监控订单状态
     */
    private function monitorTradeStatus() {
        // 开启计算刷新进程，并失败重启
        $this->startProcessAndErrorRestart(function(){
            $configTable = self::swooleTable('CONFIG');
            $requestStatusTable = self::swooleTable('API_REQUEST_RUN');
            $lastSelectTime = 0;
            while (true) {
                if(($keyExists = $configTable->exists(self::CONFIG_TABLE_KEY_LIST['TRADE_REFRESH_CONFIG_KEY'])) || time() - $lastSelectTime > self::INTERVAL_TIME_LIST['TRADE_STATUS_REFRESH']) {
                    if($keyExists) {    // 已执行，删除消息
                        $configTable->del(self::CONFIG_TABLE_KEY_LIST['TRADE_REFRESH_CONFIG_KEY']);
                    }
                    // 刷新订单信息
                    $tradeStatusList = $this->getTradeMonitorList();
                    foreach($tradeStatusList as $value) {
                        $key = $value['id'] . '_' . $value['version'];
                        $details = json_encode($value);
                        if(($record = $requestStatusTable->get($key)) && $record['details'] == $details) {
                           continue;
                        }
                        if($record) {
                            $requestStatusTable->lockFunc(function($requestStatusTable) use($key,$details){
                                if($record = $requestStatusTable->get($key)) {
                                    $record['details'] = $details;
                                } else {
                                    $record = [
                                        'last_time' => 0,
                                        'details'   => $details,
                                        'status'    => 1,
                                    ];
                                }
                                // 不存在，添加初始化
                                $requestStatusTable->set($key,$record);
                            });
                        } else {
                            // 不存在，添加初始化
                            $requestStatusTable->set($key,[
                                'last_time' => 0,
                                'details'   => $details,
                                'status'    => 1,
                            ]);
                        }
                    }
                }
                usleep(static::USLEEP_TIME_LIST['TRADE_STATUS']);
            }
        });
    }


    /**
     * 开启监控主进程
     */
    private function monitorTradeMaster() {
        $this->startProcessAndErrorRestart(function(){
            $requestStatusTable = self::swooleTable('API_REQUEST_RUN');
            while(true) {
                $nowTime = time();
                foreach($requestStatusTable as $key=>$value) {
                    $interval = null;
                    switch($value['status']) {
                        case 1: // 需请求API
                            if($tradeDetails = $requestStatusTable->lockFunc(function($requestStatusTable) use($key){
                                if(!($record = $requestStatusTable->get($key)) || $record['status'] != 1) {
                                    return false;
                                }
                                $details = json_decode($record['details'],true);
                                $requestStatusTable->set($key,array_merge($record,[
                                    'status'    => 2,
                                    'last_time' => time(),
                                ]));
                                return $details;
                            })) {
                                /*************************** 请求第三方，并获取订单状态 **************************/
                                $this->tradeProcess($tradeDetails,$key,$requestStatusTable);
                                /************************* END 请求第三方，并获取订单状态 ************************/
                            }
                            break;
                        case 2: // 已在请求中
                            $interval = self::INTERVAL_TIME_LIST['API_STATUS_WAIT'];
                            break;
                        case 3: // 处理中
                            $interval = self::INTERVAL_TIME_LIST['API_STATUS_PROCESS'];
                            break;
                        case 4: // 处理成功，但还需处理（未更新）
                            $interval = self::INTERVAL_TIME_LIST['API_STATUS_NOT_UDPATE'];
                            break;
                        case 5: // 处理成功，并进行了更新
                            $interval = self::INTERVAL_TIME_LIST['API_STATUS_FINISH'];
                            break;
                        case 6: // 处理失败，等待超时后删除
                            $interval = self::INTERVAL_TIME_LIST['API_STATUS_ERROR'];
                            break;
                    }
                    if(is_null($interval)) {
                        continue;
                    }
                    if($nowTime - $value['last_time'] >= $interval) {
                        try{
                            $requestStatusTable->del($key);
                        }catch (\Exception $e) {
                        }
                    }
                }
                usleep(self::USLEEP_TIME_LIST['TRADE_MONITOR_MASTER']);
                Swoole::closeWaitBySub();
            }
        },true);
        return true;
    }


    /**
     * 执行订单处理
     * @param $tradeDetails
     * @param $key
     */
    private function tradeProcess($tradeDetails,$key,$requestStatusTable) {
        // 开启等待线程，并执行处理
        Swoole::startWaitProcess(function($params) use($requestStatusTable) {
            list($tradeDetails,$key) = $params;
            // 获取记录数据
            if(!$record = $requestStatusTable->get($key)) {
                return false;
            }
            try {
                $requestStatusTable->set($key,array_merge($record,[
                    'status'    => 3,
                    'last_time' => time(),
                ]));
                (new Trade())->monitorStatus($tradeDetails);
                DB::close();
                $requestStatusTable->set($key,array_merge($record,[
                    'status'    => 4,
                    'last_time' => time(),
                ]));
            } catch (\Exception $e) {
                echo $e->getMessage(),"\n";
                $requestStatusTable->set($key,array_merge($record,[
                    'status'    => 6,
                    'last_time' => time(),
                ]));
            }
        },[$tradeDetails,$key],'monitorTradeMaster',self::PROCESS_CONFIG['TRADE_REQUEST_NUM']);

    }




    /**
     * 启用TCP进行数据交互 【参数：cmd params】 cmd 为指定的操作，params为需要的参数
     */
    private function createTcpServer() {
        Swoole::startProcess(function(){
            $host = '0.0.0.0';
            $port = static::tcpPort();
            $serv = new \swoole_server($host,$port);

            //swoole处理方法
            $serv->set([
                'open_eof_check' => true,   //打开EOF检测
                'package_eof'    => "\r\n", //设置EOF
            ]);


            $serv->on('connect',function($serv, $fd){
            });
            $serv->on('receive',function($serv, $fd, $reactor_id, $data){
                $data = substr($data,0,-2);
                $send = [
                    'status'    => 'error',
                    'time'  => microtime(true),
                    'param' => $data,
                    'result'  => null
                ];
                try {
                    $send['status'] = 'ok';
                    $data = explode(' ',trim($data));
                    $cmd = $data[0];
                    $params = array_key_exists(1,$data)?$data[1]:null;
                    switch(strtolower($cmd)) {
                        case 'refresh': // 刷新

                        default:
                            throw new Exception('cmd not find');
                    }

                } catch (\Exception $e) {
                    $send['status'] = 'error';
                    $send['result'] = $e->getMessage();
                }


                $serv->send($fd, json_encode($send));
            });
            $serv->on('close',function($serv, $fd){
            });
            $serv->start();
        });
    }


    /**
     * 创建及获取表数据
     * @param null $tableKey
     * @param string|null $tableSuffix
     * @return bool|mixed
     */
    private static function swooleTable($tableKey = null,string $tableSuffix = null) {
        if(is_null($tableKey)) {    // 初始化表
            foreach(static::TABLE_LIST as $table) {
                if(!$table[1]) {
                    continue;
                }
                // 创建Swoole表
                Swoole::table($table[0],$table[4],1024 * $table[3],$table[2]);
            }
            return true;
        }

        // 创建Swoole表
        $table = static::TABLE_LIST[$tableKey];
        if($tableSuffix) {
            $table[0] .= $tableSuffix;
        }
        return Swoole::table($table[0],$table[4],1024 * $table[3],$table[2]);
    }


    /**
     * 获取监控列表
     */
    private function getTradeMonitorList() {
        $result = [];
        if($list = Db::name('coin_rule_run_trade')->alias('trade')
            ->join('users_key key','key.id=trade.user_id','INNER')
            ->join('coin coin','coin.id=trade.coin_id','INNER')
            ->field([
                'trade.*',
                'key.account_id','key.access_key','key.secret_key',
                'coin.code',
            ])
            ->order([
                'trade.id' => 'DESC',
            ])
            ->where([
                ['trade.status','between',[1,3]],
                //        ['trade.request_status','<>',1],
            ])->select()) {
            $nowTime = time();
            foreach($list as $value) {
                if($value['request_status'] == 1 && $nowTime - $value['create_time'] < 60) {
                    continue;
                }
                $result[$value['id'] . '_' . $value['version']] = $value;
            }
        }
        Db::close();
        return $result;
    }



    /**
     * 监控抵扣处理
     */
    private function monitorDeduct() {
        Swoole::startProcessAndErrorRestart(function(){
            $tradeId = 0;
            while(true) {
                $where = [
                    ['trade.is_join_run','=',0],
                    ['trade.status','between',[5,6]]
                ];
                if($tradeId > 0) {
                    $where[] = ['trade.id','<',$tradeId];
                }
                $tradeList = (new CoinRuleRunTrade())->alias('trade')
                    ->join('Users_key key','trade.user_id=key.id','inner')
                    ->join('Coin_rule_run run','trade.coin_rule_run_id=run.id','inner')
                    ->limit(0,self::PROCESS_CONFIG['DEDUCT_ONE_QUERY_MAX'])
                    ->field(['trade.id','trade.order_id','trade.user_id','trade.type','trade.field_fees','trade.version',
                        'key.account_id','key.access_key','key.secret_key',
                        'run.id' => 'run_id','run.deduct_fees','run.sell_deduct_fees','run.unique_id','run.coin_rule_id','run.deduct_fees','run.version' => 'run_version'
                    ])
                    ->order('trade.id','desc')
                    ->where($where)
                    ->select();

                if(!count($tradeList)) {
                    break;
                }
                foreach($tradeList as $tradeDetails) {
                    $tradeDetails = $tradeDetails->getData();
                    $tradeId = $tradeDetails['id'];
                    try {
                        if(bccomp($tradeDetails['field_fees'],0,30) == 1) {
                            if(!($deductFees = (new CallApi())->getOrder($tradeDetails['order_id'],$tradeDetails,$isUpdate,true)) || is_null($deductFees['deduct_fees']) ) {
                                // 未获取到抵扣费用，等待下次处理
                                continue;
                            }
                            $deductFees = $deductFees['deduct_fees'];
                        } else {
                            $deductFees = 0;
                        }
                        /********************** 记录手续费处理 *****************/
                        Db::startTrans();
                        $tradeUpdate = [
                            'deduct_fees'   => $deductFees,
                            'is_join_run'   => CoinRuleRunTrade::IS_JOIN_RUN_LIST['NOT_JOIN'],
                        ];
                        // 判断是否设置为加入run表，并执行加入处理
                        if(bccomp($deductFees,0,30) != 0 && $tradeDetails['unique_id'] == strval($tradeDetails['coin_rule_id'])) {
                            $runUpdate = [];
                            $coinRuleRunModel = new CoinRuleRun();
                            switch($tradeDetails['type']) {
                                case CoinRuleRunTrade::TYPE_BUY:
                                    $runUpdate['deduct_fees'] = bcadd($tradeDetails['deduct_fees'],$deductFees,$coinRuleRunModel->scale);
                                    break;
                                case CoinRuleRunTrade::TYPE_SELL:
                                    $runUpdate['sell_deduct_fees'] = bcadd(0,$deductFees,$coinRuleRunModel->scale);
                                    break;
                            }
                            if($runUpdate) {
                                if(!$coinRuleRunModel->recordUpdate([
                                    'id'        => $tradeDetails['run_id'],
                                    'version'   => $tradeDetails['run_version'],
                                ],$runUpdate)) {
                                    throw new \Exception('run update error:' . $tradeDetails['run_id']);
                                }
                                // 加入到 run 中
                                $tradeUpdate['is_join_run'] = CoinRuleRunTrade::IS_JOIN_RUN_LIST['JOIN'];
                            }
                        }
                        // 修改交易详情
                        if(!(new CoinRuleRunTrade())->orderUpdate($tradeDetails['id'],$tradeUpdate,$tradeDetails['version'])) {
                            throw new \Exception('trade update error:' . $tradeDetails['id']);
                        }

                        Db::commit();
                        /******************** END 记录手续费处理 ***************/

                        // 添加价格偏差量整理
                        UdpServer::request(UdpServer::CMD_LIST['DEVIATIONS_CORRECT'],null,[
                            'runId' => $tradeDetails['run_id']
                        ]);

                    } catch (\Exception $e) {
                        Db::rollback();
                        // 记录sql错误信息
                        Log::write(Swoole::getErrorAndFiles($e),Log::LOG_TYPE_LIST[$e->getCode() == 400?self::PROCESS_CONFIG['DEDUCT_ERROR_LOG_TYPE']:'SQL_ERROR'],false);
                    }

                }
            }
            Db::close();
            sleep(self::INTERVAL_TIME_LIST['DEDUCT_TIME']);
        },false,self::USLEEP_TIME_LIST['ERROR_RESTART'],self::PROCESS_CONFIG['DEDUCT_MAX_REQUEST']);
    }



}